package cn.doitedu.eagle.entry;

import cn.doitedu.eagle.beans.*;
import cn.doitedu.eagle.functions.Json2BeanMapFunction;
import cn.doitedu.eagle.functions.RuleProcessFunction;
import cn.doitedu.eagle.names.OutputTags;
import cn.doitedu.eagle.names.StateDescriptors;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

import cn.doitedu.eagle.sources.KafkaSourceUtil;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/3/13
 *
 * 通过canal获取drools规则
 * 双路查询模型，drools动态规则，测试验证
 **/
public class Main {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.
                createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(2);

        String servers = "hdp01:9092,hdp02:9092,hdp03:9092";

        /**
         * 添加 事件流-->kafka source
         * **/
        DataStreamSource<String> logStream = env.
                addSource(KafkaSourceUtil.getKafkaSource(servers, "latest", "app_log"));
        //DataStreamSource<String> logStream = env.socketTextStream("localhost", 4455);

        /**
         * 添加 规则流-->kafka source
         * 并广播
         * **/
        DataStreamSource<String> ruleStream = env.addSource(KafkaSourceUtil.getKafkaSource(servers, "latest", "canal_rule"));
        MapStateDescriptor<String, RuleKieSessionStateBean> ruleStateDesc = StateDescriptors.getRuleStateDesc();
        BroadcastStream<String> ruleBroadcast = ruleStream.broadcast(ruleStateDesc);

        /**
         * json日志流，转成bean对象流
         * **/
        SingleOutputStreamOperator<LogBean> beanStream = logStream.map(new Json2BeanMapFunction());

        /**
         * 按 deviceId 分组
         * TODO 这里可以升级为动态分组
         * 整体思路：
         * 1. 这里也先接收规则流
         * 2. 从规则流中获取各规则的分组key并去重：比如放入一个set集合去重= > {k1,k2,(k3+k4),....}
         * 3. 遍历去重后的分组key，从event中用反射手段获取到key值，然后装入 Keyed包装对象
         * 4. 外部代码： keyBy(keyed->keyed.key)
         * **/
        /*beanStream
                .connect(ruleBroadcast)
                .process(new DynamicKeyFunction())
                .keyBy(keyed -> keyed.getKey());*/

        KeyedStream<LogBean, String> keyedStream = beanStream.keyBy((bean) -> bean.getDeviceId());

        /**
         * 事件流 join 广播流
         * **/
        BroadcastConnectedStream<LogBean, String> connected = keyedStream.connect(ruleBroadcast);


        // 定义一个全事件侧流输出tag
        OutputTag<LogBean> event_outs = new OutputTag<LogBean>("event_outs") {
        };

        // 定义一个 设备-规则-触发且命中次数 统计数据侧流输出tag
        OutputTag<String> deviceRuleTrigHitCountTag = OutputTags.getDeviceRuleTrigHitCountTag();

        // 定义一个触发且命中次数计数器desc
        MapStateDescriptor<String, Integer> trigHitCountDesc = StateDescriptors.getTrigHitCountDesc();

        // 定义两个缓存查询结果的：stateDesc
        MapStateDescriptor<String, String> queryCountBuffStateDesc = StateDescriptors.getQueryCountBuffStateDesc();
        MapStateDescriptor<String, String> querySeqBuffStateDesc = StateDescriptors.getQuerySeqBuffStateDesc();

        // 对查询结果缓存设置TTL
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.hours(2))
                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
                .build();
        queryCountBuffStateDesc.enableTimeToLive(ttlConfig);
        querySeqBuffStateDesc.enableTimeToLive(ttlConfig);


        // 定义一个 设备-规则-触发次数 统计数据侧流输出tag
        OutputTag<String> deviceRuleTrigCountTag = OutputTags.getDeviceRuleTrigCountTag();


        /**
         * 事件驱动，规则判断
         * 逻辑核心代码
         * **/
        SingleOutputStreamOperator<String> ruleJudged = connected.process(
                new RuleProcessFunction(
                        ruleStateDesc,
                        event_outs,
                        trigHitCountDesc,
                        deviceRuleTrigHitCountTag,
                        deviceRuleTrigCountTag,
                        queryCountBuffStateDesc,
                        querySeqBuffStateDesc
                ));

        /**
         * 打印结果
         * **/
        ruleJudged.print();
        //ruleJudged.getSideOutput(event_outs).print();
        DataStream<String> hitCounts = ruleJudged.getSideOutput(deviceRuleTrigHitCountTag);
        hitCounts.print();

        DataStream<String> trigCounts = ruleJudged.getSideOutput(deviceRuleTrigCountTag);
        //trigCounts.print();

        env.execute();
    }
}
